package com.example.springboottest.example.protobuf;


import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;


import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * Protobuf 既独立于语言，又独立于平台。Google 官方提供了多种语言的实现：Java、C#、C++、
 * GO、JavaScript 和 Python。Protobuf 数据包是一种二进制的格式，相对于文本格式的数据交换（JSON、
 * XML）来说，速度要快很多。由于 Protobuf 优异的性能，使得它更加适用于分布式应用场景下的
 * 数据通信或者异构环境下的数据交换。 与 JSON、XML 相比，Protobuf 算是后起之秀，是 Google 开源的一种数据格式。只是 Protobuf
 * 更加适合于高性能、快速响应的数据传输应用场景。另外，JSON、XML 是文本格式，数据具有可
 * 读性；而 Protobuf 是二进制数据格式，数据本身不具有可读性，只有反序列化之后才能得到真正可
 * 读的数据。正因为 Protobuf 是二进制数据格式，数据序列化之后，体积相比 JSON 和 XML 要小，
 * 更加适合网络传输。
 * proto 文件，简单地说，就是一个消息的协议文件，这个协议文件
 * 的后缀文件名为“.proto”
 * <p>
 * // [开始头部声明]
 * syntax = "proto3";
 * packagecom.crazymakercircle.netty.protocol;
 * // [结束头部声明]
 * // [开始 java 选项配置]
 * option java_package = "com.crazymakercircle.netty.protocol";
 * option java_outer_classname = "MsgProtos";
 * // [结束 java 选项配置]
 * // [开始消息定义]
 * message Msg {
 * uint32 id = 1; //消息 ID
 * string content = 2;//消息内容
 * }
 * // [结束消息定义]
 * <p>
 * <p>
 * 完成“.proto”文件定义后，下一步就是生成消息的 POJO 类和 Builder（构造者）类。有两种
 * 方式生成 Java 类：一种是通过控制台命令的方式；另一种是使用 Maven 插件的方式。
 */
public class ProtobufDemo {
    public static void main(String[] args) {
        ProtobufDemo protobufDemo = new ProtobufDemo();
        try {
            protobufDemo.serAndDesr1();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static MsgProtos.Msg buildMsg() {
        MsgProtos.Msg.Builder personBuilder = MsgProtos.Msg.newBuilder();
        personBuilder.setId(1000);
        personBuilder.setContent("疯狂创客圈:高性能学习社群");
        MsgProtos.Msg message = personBuilder.build();
        return message;
    }

    public static MsgProtos.Msg buildMsg(int id, String content) {
        MsgProtos.Msg.Builder personBuilder = MsgProtos.Msg.newBuilder();
        personBuilder.setId(id);
        personBuilder.setContent(content);
        MsgProtos.Msg message = personBuilder.build();
        return message;
    }

    //第1种方式:序列化 serialization & 反序列化 Deserialization
    public void serAndDesr1() throws IOException {
        MsgProtos.Msg message = buildMsg(1, "疯狂创客圈-高并发发烧友圈子");
        //将Protobuf对象，序列化成二进制字节数组
        byte[] data = message.toByteArray();
        //可以用于网络传输,保存到内存或外存
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        outputStream.write(data);
        data = outputStream.toByteArray();
        //二进制字节数组,反序列化成Protobuf 对象
        MsgProtos.Msg inMsg = MsgProtos.Msg.parseFrom(data);
        System.out.println("id:=" + inMsg.getId());
        System.out.println("content:=" + inMsg.getContent());
    }

    //第2种方式:序列化 serialization & 反序列化 Deserialization
    public void serAndDesr2() throws IOException {
        MsgProtos.Msg message = buildMsg();
        //序列化到二进制流
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        message.writeTo(outputStream);
        ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
        //从二进流,反序列化成Protobuf 对象
        MsgProtos.Msg inMsg = MsgProtos.Msg.parseFrom(inputStream);
        System.out.println("id:=" + inMsg.getId());
        System.out.println("content:=" + inMsg.getContent());
    }


    //第3种方式:序列化 serialization & 反序列化 Deserialization
    //带字节长度：[字节长度][字节数据],解决粘包问题
    public void serAndDesr3() throws IOException {
        MsgProtos.Msg message = buildMsg();
        //序列化到二进制流
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        message.writeDelimitedTo(outputStream);
        ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
        //从二进流,反序列化成Protobuf 对象
        MsgProtos.Msg inMsg = MsgProtos.Msg.parseDelimitedFrom(inputStream);
        System.out.println("id:=" + inMsg.getId());
        System.out.println("content:=" + inMsg.getContent());


    }

    static class MessageHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ProtoMsg.MessageNotification notification = (ProtoMsg.MessageNotification) msg;

            System.out.println("accept client notification:[" + notification.toString() + "]");

        }

    }

    private ProtoMsg.MessageNotification buildProtoBufMsg(String content) {
        ProtoMsg.MessageNotification.Builder builder = ProtoMsg.MessageNotification.newBuilder();
        builder.setMsgType(1);
        builder.setJson(content);
        return builder.build();
    }

    /**
     * 字符串解码器的使用实例
     */
    public void testProtobufDecoder() {
        ChannelInitializer i = new ChannelInitializer<EmbeddedChannel>() {
            protected void initChannel(EmbeddedChannel ch) {
                // protobufDecoder仅仅负责编码，并不支持读半包，所以在之前，一定要有读半包的处理器。
                // 有三种方式可以选择：
                // 使用netty提供ProtobufVarint32FrameDecoder
                // 继承netty提供的通用半包处理器 LengthFieldBasedFrameDecoder
                // 继承ByteToMessageDecoder类，自己处理半包

                // 半包的处理
                ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                // 需要解码的目标类
                ch.pipeline().addLast(new ProtobufDecoder(ProtoMsg.MessageNotification.getDefaultInstance()));

                ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                ch.pipeline().addLast(new ProtobufEncoder());

                ch.pipeline().addLast(new MessageHandler());
            }
        };

        EmbeddedChannel channel = new EmbeddedChannel(i);
        for (int j = 0; j < 100; j++) {

            ProtoMsg.MessageNotification pkg = buildProtoBufMsg("json " + j);

            channel.writeInbound(pkg);
            channel.flush();
        }
        try {
            Thread.sleep(Integer.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


}
